/**
 * <p>文件名:		ZookeeperClient.java</p>
 * <p>版权:		CopyrightTag</p>
 * <p>公司:		CompanyTag</p>
 * @author		周华彬(zhouhuabin@ctfo.com, zhou_hua_bin@163.com)
 */

package com.caits.lbs.framework.services.zookeeper;

import java.io.IOException;
import java.util.List;

import org.apache.logging.log4j.Logger;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;

import com.caits.lbs.framework.log.CommonLogFactory;
import com.caits.lbs.framework.services.BaseService;

/**
 * <p>
 * ZookeeperClient
 * </p>
 * <p>
 * zookeeper客户端操作类,样例，封装的东西不多
 * </p>
 *
 * @author 周华彬(zhouhuabin@ctfo.com, zhou_hua_bin@163.com)
 * @version 0.0.0
 *          <table style="border:1px solid gray;">
 *          <tr>
 *          <th width="100px">版本号</th>
 *          <th width="100px">动作</th>
 *          <th width="100px">修改人</th>
 *          <th width="100px">修改时间</th>
 *          </tr>
 *          <!-- 以 Table 方式书写修改历史 -->
 *          <tr>
 *          <td>0.0.0</td>
 *          <td>创建类</td>
 *          <td>zhouhuabin</td>
 *          <td>2015年4月17日 下午2:50:12</td>
 *          </tr>
 *          <tr>
 *          <td>XXX</td>
 *          <td>XXX</td>
 *          <td>XXX</td>
 *          <td>XXX</td>
 *          </tr>
 *          </table>
 */
public class ZookeeperClient implements BaseService{
	
	/** 变量:TODO,类型:int */
	private final int CONNECTION_TIMEOUT;
	/** 日志记录器 */
	private Logger log = CommonLogFactory.getLog();
	/** 服务器地址,例如"127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002 */
	private String serverAddress = "redis-shard3:2181";
	/** 连接对象 */
	private ZooKeeper zkClient;

	
	/**
	 * 构造器 
	 * @param serverPairs	例如"127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002
	 * @param watcher	监视器，数据变更时回调
	 */
	public  ZookeeperClient(String serverPairs,int timeout,Watcher watcher) {
		serverAddress=serverPairs;
		CONNECTION_TIMEOUT = timeout;
		try {
			zkClient = new ZooKeeper(serverAddress,CONNECTION_TIMEOUT , watcher);
		} catch (IOException e) {
			log.error("zookeeper调用发生异常", e);
		}
	}
	/**
	 * /consumers/[group_id]/offsets/[topic]/[broker_id-partition_id]   -->offset_value
	 * 根据path获取值 
	 * @param path
	 * @return
	 */
	public byte[] getData(String path) {
		try {
			return zkClient.getData(path, true, null);
		} catch (KeeperException e) {
			log.error("zookeeper调用发生异常", e);
		} catch (InterruptedException e) {
			log.error("zookeeper调用发生异常", e);
		}
		return new byte[0];
	}
	/**
	 * /consumers/[group_id]/offsets/[topic]/[broker_id-partition_id]   -->offset_value
	 * 修改指定路径的数据 
	 * @param path
	 * @param val
	 */
	public void setData(String path,byte[] val) {
		try {
			Stat stat =  zkClient.setData(path, val, -1);
			log.info("数据状态getAversion="+stat.getAversion());
		} catch (KeeperException e) {
			log.error("zookeeper调用发生异常", e);
		} catch (InterruptedException e) {
			log.error("zookeeper调用发生异常", e);
		}
	}
	/**
	 * 
	 * 返回目录列表 
	 * @param path
	 * @return
	 */
	public List<String> getChildren(String path) {
		try {
			return zkClient.getChildren(path, true);
		} catch (KeeperException e) {
			log.error("zookeeper调用发生异常", e);
		} catch (InterruptedException e) {
			log.error("zookeeper调用发生异常", e);
		}
		return null;
	}
	/**
	 * 仅仅是测试方法，创建一些目录再删除 
	 */
	@Deprecated
	public void test1() {
		try {
//			//consumers/[group_id]/offsets/[topic]/[broker_id-partition_id]   -->offset_value
			// 创建一个目录节点
			zkClient.create("/testRootPath", "testRootData".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
			// 创建一个子目录节点
			zkClient.create("/testRootPath/testChildPathOne", "testChildDataOne".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
			log.info(new String(zkClient.getData("/testRootPath", false, null)));
			// 取出子目录节点列表
			log.info(zkClient.getChildren("/testRootPath", true));
			// 修改子目录节点数据
			zkClient.setData("/testRootPath/testChildPathOne", "modifyChildDataOne".getBytes(), -1);
			log.info("目录节点状态：[" + zkClient.exists("/testRootPath", true) + "]");
			// 创建另外一个子目录节点
			zkClient.create("/testRootPath/testChildPathTwo", "testChildDataTwo".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
			log.info(new String(zkClient.getData("/testRootPath/testChildPathTwo", true, null)));
			// 删除子目录节点
			zkClient.delete("/testRootPath/testChildPathTwo", -1);
			zkClient.delete("/testRootPath/testChildPathOne", -1);
			// 删除父目录节点
			zkClient.delete("/testRootPath", -1);
		} catch (Exception e) {
			log.error("zookeeper调用发生异常", e);
		}
	}
	/**
	 * 获得zookeeper连接对象 
	 * @return
	 */
	public ZooKeeper getZkClient(){
		return zkClient;
	} 

	/**
	 * 关闭client连接 
	 */
	public void close() {
		try {
			zkClient.close();
		} catch (Exception e) {
			log.error("zookeeper调用发生异常", e);
		}
	}
	@Override
	public void resetParameter() {
		// TODO Auto-generated method stub
		
	}
}
